package com.catmiao.spark.stream

import com.catmiao.spark.util.JDBCUtil
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.sql.ResultSet
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

/**
 * @title: SparkStreaming01_WordCount
 * @projectName spark_study
 * @description: 用户黑名单
 * @author ChengMiao
 * @date 2024/3/25 00:31
 */
object SparkStreaming12_Req1_BlackList {

  def main(args: Array[String]): Unit = {


    // 创建环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    //  param1 : 环境配置，SparkConf
    //  param2 ： 采集周期【批量处理周期】
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    //3.定义 Kafka 参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG ->
        "localhost:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "test",
      "key.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //4.读取 Kafka 数据创建 DStream
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String](ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Set("first"), kafkaPara))


    val dataDs: DStream[AdClickData] = kafkaDStream.map(
      data => {
        val d: String = data.value()
        val value = d.split(" ")
        AdClickData(value(0), value(1), value(2), value(3), value(4))
      }
    )


    val ds: DStream[((String, String, String), Int)] = dataDs.transform(
      rdd => {
        // TODO 周期性获取黑名单数据 JDBC

        val blackList = ListBuffer[String]()
        val conn = JDBCUtil.getConnection

        val preparedStatement = conn.prepareStatement("select userid from black_list")

        val rs: ResultSet = preparedStatement.executeQuery()
        while (rs.next()) {
          blackList.append(rs.getString(1))
        }

        rs.close()
        preparedStatement.close()
        conn.close()

        // TODO 判断点击用户是否在黑名单中
        val filterRdd: RDD[AdClickData] = rdd.filter(
          data => {
            !blackList.contains(data.user)
          }
        )

        // TODO 如果用户不在黑名单中，进行统计数量【每一个采集周期】
        val ds = filterRdd.map(
          data => {
            val day = new SimpleDateFormat("yyyy-MM-dd").format(new Date(data.ts.toLong))
            val user = data.user
            val ad = data.ad
            ((day, user, ad), 1)
          }
        ).reduceByKey(_ + _)
        ds
      }
    )


    ds.foreachRDD(
      rdd => {
        rdd.foreach {
          case ((day, user, ad), count) => {

            println(s"${day} ${user} ${ad} ${count}")
            if (count >= 30) {
              // TODO 如果统计数量超过点击阈值，将用户拉入黑名单
              val conn = JDBCUtil.getConnection

              val preparedStatement = conn.prepareStatement(
                """
                  | insert into black_list(userid) values (?)
                  | on DUPLICATE KEY
                  | update userid = ?
              """.stripMargin)
              preparedStatement.setString(1, user)
              preparedStatement.setString(2, user)

              preparedStatement.executeUpdate()

              preparedStatement.close()
              conn.close()
            } else {
              // TODO 如果没有超过阈值，将当天广告的点击数量继续努更新
              val conn = JDBCUtil.getConnection

              val preparedStatement = conn.prepareStatement(
                """
                  | select * from user_ad_count
                  | where dt = ? and userid = ? and adid = ?
                """.stripMargin)
              preparedStatement.setString(1, day)
              preparedStatement.setString(2, user)
              preparedStatement.setString(3, ad)

              val rs: ResultSet = preparedStatement.executeQuery()
              // 查询统计表数据
              if(rs.next()){
                // 如果存在则更新
                val preparedStatement1 = conn.prepareStatement(
                  """
                    | update user_ad_count
                    | set count = count + ?
                    | where dt = ? and userid = ? and adid = ?
                  """.stripMargin)
                preparedStatement1.setInt(1 ,count)
                preparedStatement1.setString(2, day)
                preparedStatement1.setString(3, user)
                preparedStatement1.setString(4, ad)

                preparedStatement1.executeUpdate()
                preparedStatement1.close()

                // TODO 更新后的点击数据是否超过阈值，如果超过，将用户拉入到黑名单
                val preparedStatement2 = conn.prepareStatement(
                  """
                    | select * from user_ad_count
                    | where dt = ? and userid = ? and adid = ? and count >=30
  """.stripMargin)
                preparedStatement2.setString(1, day)
                preparedStatement2.setString(2, user)
                preparedStatement2.setString(3, ad)
                val rs2: ResultSet = preparedStatement2.executeQuery()
                if(rs2.next()){
                  // 如果超过，将用户拉入到黑名单
                  val preparedStatement3 = conn.prepareStatement(
                    """
                      | insert into black_list(userid) values (?)
                      | on DUPLICATE KEY
                      | update userid = ?
                  """.stripMargin)
                  preparedStatement3.setString(1, user)
                  preparedStatement3.setString(2, user)

                  preparedStatement3.executeUpdate()

                  preparedStatement3.close()
                }
                rs2.close()
                preparedStatement2.close()
              }else {
                // 不存在则新增

                val preparedStatement1 = conn.prepareStatement(
                  """
                    | insert into user_ad_count(dt,userid,adid,count) values (?,?,?,?)
                  """.stripMargin)
                preparedStatement1.setString(1, day)
                preparedStatement1.setString(2, user)
                preparedStatement1.setString(3, ad)
                preparedStatement1.setInt(4, count)

                preparedStatement1.executeUpdate()
                preparedStatement1.close()
              }

              rs.close()
              preparedStatement.close()
              conn.close()

            }
          }
        }
      }
    )


    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 广告点击数据
   *
   * @param ts
   * @param area
   * @param city
   * @param user
   * @param ad
   */
  case class AdClickData(ts: String, area: String, city: String, user: String, ad: String)


}
